package cloudy.bolt;

import cloudy.hbase.dao.HBaseDAO;
import cloudy.hbase.dao.impl.HBaseDAOImpl;
import cloudy.tools.DateFmt;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.kafka.common.utils.ByteUtils;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.IBasicBolt;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.math.BigDecimal;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Vector;

/**
 * @Name cloudy.bolt.AreaAmtBolt
 * @Description
 * @Author Elwyn
 * @Version 2017/9/29
 * @Copyright 上海云辰信息科技有限公司
 **/
public class AreaAmtBolt implements IBasicBolt {
	private Map<String, Double> countMap;
	private HBaseDAO hBaseDAO;
	private String today;

	@Override
	public void prepare(Map map, TopologyContext topologyContext) {
		today = DateFmt.getCountDate(null, DateFmt.date_short);
		//初始化
		hBaseDAO = new HBaseDAOImpl();
		countMap = this.initMap(today, hBaseDAO);


	}

	@Override
	public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
		if (tuple != null) {
			String area_id = tuple.getString(0);
			Double order_amt = Double.valueOf(tuple.getString(1));
			String order_date = tuple.getStringByField("order_date");
			if (!order_date.equals(today)){
				countMap.clear();
			}
			Double count = countMap.get(order_date + "_" + area_id);
			if (count == null) {
				count = 0.0;
			}
			count += order_amt;
			countMap.put(order_date + "_" + area_id, count);
			System.err.println("areaAmtBolt:" + order_date + "_" + area_id + "count:" + count);
			basicOutputCollector.emit(new Values(order_date + "_" + area_id, count));
		}
	}

	@Override
	public void cleanup() {
		countMap.clear();
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
		outputFieldsDeclarer.declare(new Fields("date_area", "amt"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		return null;
	}

	private Map<String, Double> initMap(String rowKeyDate, HBaseDAO hBaseDAO) {
		Map<String, Double> countMap = new HashMap<>();
		List<Result> resultList = hBaseDAO.getRows("area_order", rowKeyDate, new String[]{"order_amt"});

		for (Result result : resultList) {
			for (Cell cell : result.listCells()) {
				String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
				if ("order_amt".equals(rowKey)) {
					countMap.put(rowKey, Bytes.toDouble(CellUtil.cloneValue(cell)));
					break;
				}
			}
		}
		return countMap;
	}
}
